package k8s

import (
	"context"
	"sync"

	"gitee.com/mkwchecking/sr-admin/pkg/eventbus"
	"k8s.io/apimachinery/pkg/watch"
)

type SourceWatchCallback[T any] func(SourceWatchMsg[T])

type SourceWatchMsg[T any] struct {
	Key       string
	Source    T
	EventType watch.EventType
}

type sourceWatcher[S any] struct {
	lock sync.Mutex
}

func (s *sourceWatcher[S]) AddSubscribers(key string, cb ...SourceWatchCallback[S]) {
	for _, c := range cb {
		eventbus.EventSubscribe(key, c)
	}
}

func (s sourceWatcher[S]) GoWatch(ctx context.Context, key string, ch watch.Interface, cb ...SourceWatchCallback[S]) {
	s.lock.Lock()
	defer s.lock.Unlock()
	for _, c := range cb {
		eventbus.EventSubscribe(key, c)
	}

	go func() {
		for {
			select {
			case <-ctx.Done():
				ch.Stop()
				return
			case event := <-ch.ResultChan():
				data := event.Object.DeepCopyObject()

				// 发布事件
				eventbus.EventPublish(key, &SourceWatchMsg[S]{
					Key:       key,
					Source:    data.(S),
					EventType: event.Type,
				})

			}
		}
	}()
}

func (s *sourceWatcher[S]) Unwatched(ctx context.Context, key string, cb ...SourceWatchCallback[S]) {
	s.lock.Lock()
	defer s.lock.Unlock()

	ctx.Done()
	for _, c := range cb {
		eventbus.EventUnsubscribe(key, c)
	}
}

func AddSubscribers[T any](key string, cb ...SourceWatchCallback[T]) error {
	for _, c := range cb {
		if err := eventbus.EventSubscribe(key, c); err != nil {
			return err
		}
	}
	return nil
}
